/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.apache.iotdb;

import org.apache.iotdb.isession.SessionDataSet;
import org.apache.iotdb.isession.template.Template;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.Session;
import org.apache.iotdb.session.template.InternalNode;
import org.apache.iotdb.session.template.MeasurementNode;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.utils.BitMap;
import org.apache.iotdb.tsfile.write.record.Tablet;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;

import java.io.IOException;
import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@SuppressWarnings("squid:S106")
public class AlignedTimeseriesSessionExample {

    private static Session session;
    private static final String ROOT_SG1_D1 = "root.sg_1.d1";
    private static final String ROOT_SG1_D1_VECTOR2 = "root.sg_1.d1.vector2";
    private static final String ROOT_SG1_D1_VECTOR3 = "root.sg_1.d1.vector3";
    private static final String ROOT_SG2_D1_VECTOR4 = "root.sg_2.d1.vector4";
    private static final String ROOT_SG2_D1_VECTOR5 = "root.sg_2.d1.vector5";
    private static final String ROOT_SG2_D1_VECTOR6 = "root.sg_2.d1.vector6";
    private static final String ROOT_SG2_D1_VECTOR7 = "root.sg_2.d1.vector7";
    private static final String ROOT_SG2_D1_VECTOR8 = "root.sg_2.d1.vector8";

    public static void main(String[] args)
            throws IoTDBConnectionException, StatementExecutionException {
        session = new Session("127.0.0.1", 6667, "root", "root");
        session.open(false);

        // set session fetchSize
        session.setFetchSize(10000);

        //    createTemplate();
        createAlignedTimeseries();

        insertAlignedRecord();
        //    insertAlignedRecords();
        //    insertAlignedRecordsOfOneDevice();

        //    insertAlignedStringRecord();
        //    insertAlignedStringRecords();

        //    insertTabletWithAlignedTimeseriesMethod1();
        //    insertTabletWithAlignedTimeseriesMethod2();
        //    insertNullableTabletWithAlignedTimeseries();
        //    insertTabletsWithAlignedTimeseries();
        session.executeNonQueryStatement("flush");
        selectTest();
        selectWithValueFilterTest();
        selectWithLastTest();
        selectWithLastTestWithoutValueFilter();
        session.executeNonQueryStatement("delete from root.sg_1.d1.s1 where time <= 5");
        System.out.println("execute sql delete from root.sg_1.d1.s1 where time <= 5");
        selectTest();
        selectWithValueFilterTest();
        selectWithLastTest();
        selectWithLastTestWithoutValueFilter();
        session.executeNonQueryStatement("delete from root.sg_1.d1.s2 where time <= 3");
        System.out.println("execute sql delete from root.sg_1.d1.s2 where time <= 3");

        selectTest();
        selectWithValueFilterTest();
        selectWithLastTest();
        selectWithLastTestWithoutValueFilter();
        session.executeNonQueryStatement("delete from root.sg_1.d1.s1 where time <= 10");
        System.out.println("execute sql delete from root.sg_1.d1.s1 where time <= 10");
        selectTest();
        selectWithValueFilterTest();
        selectWithLastTest();
        selectWithLastTestWithoutValueFilter();

        //    selectWithValueFilterTest();
        //    selectWithGroupByTest();
        //    selectWithLastTest();

        //    selectWithAggregationTest();

        //    selectWithAlignByDeviceTest();

        session.close();
    }

    private static void selectTest() throws StatementExecutionException, IoTDBConnectionException {
        SessionDataSet dataSet = session.executeQueryStatement("select s1 from root.sg_1.d1");
        System.out.println(dataSet.getColumnNames());
        while (dataSet.hasNext()) {
            System.out.println(dataSet.next());
        }

        dataSet.closeOperationHandle();
        dataSet = session.executeQueryStatement("select * from root.sg_1.d1");
        System.out.println(dataSet.getColumnNames());
        while (dataSet.hasNext()) {
            System.out.println(dataSet.next());
        }

        dataSet.closeOperationHandle();
    }

    private static void selectWithAlignByDeviceTest()
            throws StatementExecutionException, IoTDBConnectionException {
        SessionDataSet dataSet =
                session.executeQueryStatement("select * from root.sg_1 align by device");
        System.out.println(dataSet.getColumnNames());
        while (dataSet.hasNext()) {
            System.out.println(dataSet.next());
        }

        dataSet.closeOperationHandle();
    }

    private static void selectWithValueFilterTest()
            throws StatementExecutionException, IoTDBConnectionException {
        SessionDataSet dataSet =
                session.executeQueryStatement("select s1 from root.sg_1.d1 where s1 > 3 and time < 9");
        System.out.println(dataSet.getColumnNames());
        while (dataSet.hasNext()) {
            System.out.println(dataSet.next());
        }

        dataSet.closeOperationHandle();
        dataSet =
                session.executeQueryStatement(
                        "select * from root.sg_1.d1 where time < 8 and s1 > 3 and s2 > 5");
        System.out.println(dataSet.getColumnNames());
        while (dataSet.hasNext()) {
            System.out.println(dataSet.next());
        }

        dataSet.closeOperationHandle();
    }

    private static void selectWithAggregationTest()
            throws StatementExecutionException, IoTDBConnectionException {
        SessionDataSet dataSet = session.executeQueryStatement("select count(s1) from root.sg_1.d1");
        System.out.println(dataSet.getColumnNames());
        while (dataSet.hasNext()) {
            System.out.println(dataSet.next());
        }

        dataSet.closeOperationHandle();
        dataSet = session.executeQueryStatement("select count(*) from root.sg_1.d1");
        System.out.println(dataSet.getColumnNames());
        while (dataSet.hasNext()) {
            System.out.println(dataSet.next());
        }

        dataSet.closeOperationHandle();
        dataSet =
                session.executeQueryStatement(
                        "select sum(*) from root.sg_1.d1.vector where time > 50 and s1 > 0 and s2 > 10000");
        System.out.println(dataSet.getColumnNames());
        while (dataSet.hasNext()) {
            System.out.println(dataSet.next());
        }

        dataSet.closeOperationHandle();
    }

    private static void selectWithGroupByTest()
            throws StatementExecutionException, IoTDBConnectionException {
        SessionDataSet dataSet =
                session.executeQueryStatement(
                        "select count(s1) from root.sg_1.d1.vector GROUP BY ([1, 100), 20ms)");
        System.out.println(dataSet.getColumnNames());
        while (dataSet.hasNext()) {
            System.out.println(dataSet.next());
        }

        dataSet.closeOperationHandle();
        dataSet =
                session.executeQueryStatement(
                        "select count(*) from root.sg_1.d1.vector where time > 50 and s1 > 0 and s2 > 10000"
                                + " GROUP BY ([50, 100), 10ms)");
        System.out.println(dataSet.getColumnNames());
        while (dataSet.hasNext()) {
            System.out.println(dataSet.next());
        }

        dataSet.closeOperationHandle();
    }

    private static void selectWithLastTest()
            throws StatementExecutionException, IoTDBConnectionException {
        SessionDataSet dataSet = session.executeQueryStatement("select last s1 from root.sg_1.d1");
        System.out.println(dataSet.getColumnNames());
        while (dataSet.hasNext()) {
            System.out.println(dataSet.next());
        }

        dataSet.closeOperationHandle();
        dataSet = session.executeQueryStatement("select last * from root.sg_1.d1");
        System.out.println(dataSet.getColumnNames());
        while (dataSet.hasNext()) {
            System.out.println(dataSet.next());
        }

        dataSet.closeOperationHandle();
    }

    private static void selectWithLastTestWithoutValueFilter()
            throws StatementExecutionException, IoTDBConnectionException {
        SessionDataSet dataSet =
                session.executeQueryStatement("select last s1 from root.sg_1.d1 where time >= 5");
        System.out.println(dataSet.getColumnNames());
        while (dataSet.hasNext()) {
            System.out.println(dataSet.next());
        }

        dataSet.closeOperationHandle();

        dataSet = session.executeQueryStatement("select last * from root.sg_1.d1 where time >= 5");
        System.out.println(dataSet.getColumnNames());
        while (dataSet.hasNext()) {
            System.out.println(dataSet.next());
        }
        dataSet.closeOperationHandle();

        dataSet = session.executeQueryStatement("select last * from root.sg_1.d1 where time >= 20");
        System.out.println(dataSet.getColumnNames());
        while (dataSet.hasNext()) {
            System.out.println(dataSet.next());
        }
        dataSet.closeOperationHandle();
    }

    private static void createAlignedTimeseries()
            throws StatementExecutionException, IoTDBConnectionException {
        List<String> measurements = new ArrayList<>();
        for (int i = 1; i <= 2; i++) {
            measurements.add("s" + i);
        }
        List<TSDataType> dataTypes = new ArrayList<>();
        dataTypes.add(TSDataType.INT64);
        dataTypes.add(TSDataType.INT32);
        List<TSEncoding> encodings = new ArrayList<>();
        List<CompressionType> compressors = new ArrayList<>();
        for (int i = 1; i <= 2; i++) {
            encodings.add(TSEncoding.RLE);
            compressors.add(CompressionType.SNAPPY);
        }
        session.createAlignedTimeseries(
                ROOT_SG1_D1, measurements, dataTypes, encodings, compressors, null, null, null);
    }

    // be sure template is coordinate with tablet
    private static void createTemplate()
            throws StatementExecutionException, IoTDBConnectionException, IOException {
        Template template = new Template("template1");
        InternalNode iNodeVector = new InternalNode("vector", true);
        MeasurementNode mNodeS1 =
                new MeasurementNode("s1", TSDataType.INT64, TSEncoding.RLE, CompressionType.SNAPPY);
        MeasurementNode mNodeS2 =
                new MeasurementNode("s2", TSDataType.INT32, TSEncoding.RLE, CompressionType.SNAPPY);

        iNodeVector.addChild(mNodeS1);
        iNodeVector.addChild(mNodeS2);

        template.addToTemplate(iNodeVector);

        session.createSchemaTemplate(template);
        session.setSchemaTemplate("template1", "root.sg_1");
    }

    /**
     * Method 1 for insert tablet with aligned timeseries
     */
    private static void insertTabletWithAlignedTimeseriesMethod1()
            throws IoTDBConnectionException, StatementExecutionException {
        // The schema of measurements of one device
        // only measurementId and data type in MeasurementSchema take effects in Tablet
        List<MeasurementSchema> schemaList = new ArrayList<>();
        schemaList.add(new MeasurementSchema("s1", TSDataType.INT64));
        schemaList.add(new MeasurementSchema("s2", TSDataType.INT32));

        Tablet tablet = new Tablet(ROOT_SG1_D1, schemaList);
        long timestamp = 1;

        for (long row = 1; row < 100; row++) {
            int rowIndex = tablet.rowSize++;
            tablet.addTimestamp(rowIndex, timestamp);
            tablet.addValue(
                    schemaList.get(0).getMeasurementId(), rowIndex, new SecureRandom().nextLong());
            tablet.addValue(schemaList.get(1).getMeasurementId(), rowIndex, new SecureRandom().nextInt());

            if (tablet.rowSize == tablet.getMaxRowNumber()) {
                session.insertAlignedTablet(tablet, true);
                tablet.reset();
            }
            timestamp++;
        }

        if (tablet.rowSize != 0) {
            session.insertAlignedTablet(tablet);
            tablet.reset();
        }

        session.executeNonQueryStatement("flush");
    }

    /**
     * Method 2 for insert tablet with aligned timeseries
     */
    private static void insertTabletWithAlignedTimeseriesMethod2()
            throws IoTDBConnectionException, StatementExecutionException {
        // The schema of measurements of one device
        // only measurementId and data type in MeasurementSchema take effects in Tablet
        List<MeasurementSchema> schemaList = new ArrayList<>();
        schemaList.add(new MeasurementSchema("s1", TSDataType.INT64));
        schemaList.add(new MeasurementSchema("s2", TSDataType.INT32));

        Tablet tablet = new Tablet(ROOT_SG1_D1_VECTOR2, schemaList);
        long[] timestamps = tablet.timestamps;
        Object[] values = tablet.values;

        for (long time = 100; time < 200; time++) {
            int row = tablet.rowSize++;
            timestamps[row] = time;

            long[] sensor1 = (long[]) values[0];
            sensor1[row] = new SecureRandom().nextLong();

            int[] sensor2 = (int[]) values[1];
            sensor2[row] = new SecureRandom().nextInt();

            if (tablet.rowSize == tablet.getMaxRowNumber()) {
                session.insertAlignedTablet(tablet, true);
                tablet.reset();
            }
        }

        if (tablet.rowSize != 0) {
            session.insertAlignedTablet(tablet, true);
            tablet.reset();
        }

        session.executeNonQueryStatement("flush");
    }

    private static void insertNullableTabletWithAlignedTimeseries()
            throws IoTDBConnectionException, StatementExecutionException {
        // The schema of measurements of one device
        // only measurementId and data type in MeasurementSchema take effects in Tablet
        List<MeasurementSchema> schemaList = new ArrayList<>();
        schemaList.add(new MeasurementSchema("s1", TSDataType.INT64));
        schemaList.add(new MeasurementSchema("s2", TSDataType.INT32));

        Tablet tablet = new Tablet(ROOT_SG1_D1_VECTOR3, schemaList);

        long[] timestamps = tablet.timestamps;
        Object[] values = tablet.values;
        // Use the bitMap to mark the null value point
        BitMap[] bitMaps = new BitMap[values.length];
        tablet.bitMaps = bitMaps;

        bitMaps[1] = new BitMap(tablet.getMaxRowNumber());
        for (long time = 200; time < 300; time++) {
            int row = tablet.rowSize++;
            timestamps[row] = time;

            long[] sensor1 = (long[]) values[0];
            sensor1[row] = new SecureRandom().nextLong();

            int[] sensor2 = (int[]) values[1];
            sensor2[row] = new SecureRandom().nextInt();

            // mark this point as null value
            if (time % 5 == 0) {
                bitMaps[1].mark(row);
            }

            if (tablet.rowSize == tablet.getMaxRowNumber()) {
                session.insertAlignedTablet(tablet, true);
                tablet.reset();
                bitMaps[1].reset();
            }
        }

        if (tablet.rowSize != 0) {
            session.insertAlignedTablet(tablet, true);
            tablet.reset();
        }

        session.executeNonQueryStatement("flush");
    }

    private static void insertAlignedRecord()
            throws IoTDBConnectionException, StatementExecutionException {
        // first file we have both sensots' data
        List<String> measurements = new ArrayList<>();
        List<TSDataType> types = new ArrayList<>();
        measurements.add("s1");
        measurements.add("s2");
        types.add(TSDataType.INT64);
        types.add(TSDataType.INT32);

        for (long time = 0; time < 10; time++) {
            List<Object> values = new ArrayList<>();
            values.add(time);
            values.add((int) time);
            session.insertAlignedRecord(ROOT_SG1_D1, time, measurements, types, values);
        }
        session.executeNonQueryStatement("flush");
        // second file we only have s1's data
        measurements.clear();
        types.clear();
        measurements.add("s1");
        types.add(TSDataType.INT64);
        for (long time = 10; time < 20; time++) {
            List<Object> values = new ArrayList<>();
            values.add(time);
            session.insertAlignedRecord(ROOT_SG1_D1, time, measurements, types, values);
        }
    }

    private static void insertAlignedStringRecord()
            throws IoTDBConnectionException, StatementExecutionException {
        List<String> measurements = new ArrayList<>();
        measurements.add("s1");
        measurements.add("s2");

        for (long time = 0; time < 1; time++) {
            List<String> values = new ArrayList<>();
            values.add("3");
            values.add("4");
            session.insertAlignedRecord(ROOT_SG2_D1_VECTOR5, time, measurements, values);
        }
    }

    private static void insertAlignedRecords()
            throws IoTDBConnectionException, StatementExecutionException {
        List<String> deviceIds = new ArrayList<>();
        List<List<String>> measurementsList = new ArrayList<>();
        List<List<TSDataType>> typeList = new ArrayList<>();
        List<Long> times = new ArrayList<>();
        List<List<Object>> valueList = new ArrayList<>();

        for (long time = 1; time < 5; time++) {
            List<String> measurements = new ArrayList<>();
            measurements.add("s1");
            measurements.add("s2");

            List<TSDataType> types = new ArrayList<>();
            types.add(TSDataType.INT64);
            types.add(TSDataType.INT32);

            List<Object> values = new ArrayList<>();
            values.add(1L);
            values.add(2);

            deviceIds.add(ROOT_SG2_D1_VECTOR4);
            times.add(time);
            measurementsList.add(measurements);
            typeList.add(types);
            valueList.add(values);
        }
        session.insertAlignedRecords(deviceIds, times, measurementsList, typeList, valueList);
    }

    private static void insertAlignedStringRecords()
            throws IoTDBConnectionException, StatementExecutionException {
        List<String> deviceIds = new ArrayList<>();
        List<List<String>> measurementsList = new ArrayList<>();
        List<Long> times = new ArrayList<>();
        List<List<String>> valueList = new ArrayList<>();

        for (long time = 1; time < 5; time++) {
            List<String> measurements = new ArrayList<>();
            measurements.add("s1");
            measurements.add("s2");

            List<String> values = new ArrayList<>();
            values.add("3");
            values.add("4");

            deviceIds.add(ROOT_SG2_D1_VECTOR5);
            times.add(time);
            measurementsList.add(measurements);
            valueList.add(values);
        }
        session.insertAlignedRecords(deviceIds, times, measurementsList, valueList);
    }

    private static void insertAlignedRecordsOfOneDevice()
            throws IoTDBConnectionException, StatementExecutionException {
        List<List<String>> measurementsList = new ArrayList<>();
        List<List<TSDataType>> typeList = new ArrayList<>();
        List<Long> times = new ArrayList<>();
        List<List<Object>> valueList = new ArrayList<>();

        for (long time = 10; time < 15; time++) {
            List<String> measurements = new ArrayList<>();
            measurements.add("s1");
            measurements.add("s2");

            List<TSDataType> types = new ArrayList<>();
            types.add(TSDataType.INT64);
            types.add(TSDataType.INT32);

            List<Object> values = new ArrayList<>();
            values.add(1L);
            values.add(2);

            times.add(time);
            measurementsList.add(measurements);
            typeList.add(types);
            valueList.add(values);
        }
        session.insertAlignedRecordsOfOneDevice(
                ROOT_SG2_D1_VECTOR4, times, measurementsList, typeList, valueList);
    }

    private static void insertTabletsWithAlignedTimeseries()
            throws IoTDBConnectionException, StatementExecutionException {

        List<MeasurementSchema> schemaList1 = new ArrayList<>();
        schemaList1.add(new MeasurementSchema("s1", TSDataType.INT64));
        schemaList1.add(new MeasurementSchema("s2", TSDataType.INT64));

        List<MeasurementSchema> schemaList2 = new ArrayList<>();
        schemaList2.add(new MeasurementSchema("s1", TSDataType.INT64));
        schemaList2.add(new MeasurementSchema("s2", TSDataType.INT64));

        List<MeasurementSchema> schemaList3 = new ArrayList<>();
        schemaList3.add(new MeasurementSchema("s1", TSDataType.INT64));
        schemaList3.add(new MeasurementSchema("s2", TSDataType.INT64));

        Tablet tablet1 = new Tablet(ROOT_SG2_D1_VECTOR6, schemaList1, 100);
        Tablet tablet2 = new Tablet(ROOT_SG2_D1_VECTOR7, schemaList2, 100);
        Tablet tablet3 = new Tablet(ROOT_SG2_D1_VECTOR8, schemaList3, 100);

        Map<String, Tablet> tabletMap = new HashMap<>();
        tabletMap.put(ROOT_SG2_D1_VECTOR6, tablet1);
        tabletMap.put(ROOT_SG2_D1_VECTOR7, tablet2);
        tabletMap.put(ROOT_SG2_D1_VECTOR8, tablet3);

        // Method 1 to add tablet data
        long timestamp = System.currentTimeMillis();
        for (long row = 0; row < 100; row++) {
            int row1 = tablet1.rowSize++;
            int row2 = tablet2.rowSize++;
            int row3 = tablet3.rowSize++;
            tablet1.addTimestamp(row1, timestamp);
            tablet2.addTimestamp(row2, timestamp);
            tablet3.addTimestamp(row3, timestamp);
            for (int i = 0; i < 2; i++) {
                long value = new SecureRandom().nextLong();
                tablet1.addValue(schemaList1.get(i).getMeasurementId(), row1, value);
                tablet2.addValue(schemaList2.get(i).getMeasurementId(), row2, value);
                tablet3.addValue(schemaList3.get(i).getMeasurementId(), row3, value);
            }
            if (tablet1.rowSize == tablet1.getMaxRowNumber()) {
                session.insertAlignedTablets(tabletMap, true);
                tablet1.reset();
                tablet2.reset();
                tablet3.reset();
            }
            timestamp++;
        }

        if (tablet1.rowSize != 0) {
            session.insertAlignedTablets(tabletMap, true);
            tablet1.reset();
            tablet2.reset();
            tablet3.reset();
        }

        // Method 2 to add tablet data
        long[] timestamps1 = tablet1.timestamps;
        Object[] values1 = tablet1.values;
        long[] timestamps2 = tablet2.timestamps;
        Object[] values2 = tablet2.values;
        long[] timestamps3 = tablet3.timestamps;
        Object[] values3 = tablet3.values;

        for (long time = 0; time < 100; time++) {
            int row1 = tablet1.rowSize++;
            int row2 = tablet2.rowSize++;
            int row3 = tablet3.rowSize++;
            timestamps1[row1] = time;
            timestamps2[row2] = time;
            timestamps3[row3] = time;
            for (int i = 0; i < 2; i++) {
                long[] sensor1 = (long[]) values1[i];
                sensor1[row1] = i;
                long[] sensor2 = (long[]) values2[i];
                sensor2[row2] = i;
                long[] sensor3 = (long[]) values3[i];
                sensor3[row3] = i;
            }
            if (tablet1.rowSize == tablet1.getMaxRowNumber()) {
                session.insertAlignedTablets(tabletMap, true);

                tablet1.reset();
                tablet2.reset();
                tablet3.reset();
            }
        }

        if (tablet1.rowSize != 0) {
            session.insertAlignedTablets(tabletMap, true);
            tablet1.reset();
            tablet2.reset();
            tablet3.reset();
        }
    }
}
